package xyz.riceball.springcloudstreambinderredis.component;

import org.redisson.api.RedissonClient;
import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.integration.core.MessageProducer;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import xyz.riceball.springcloudstreambinderredis.property.RedisConsumerProperties;
import xyz.riceball.springcloudstreambinderredis.property.RedisProducerProperties;

/**
 * <p>
 * RedisMessageChannelBinder
 * </p>
 *
 * @author xiaovcloud
 * @since 2024/6/24 16:37
 */
public class RedisMessageChannelBinder extends AbstractMessageChannelBinder<
	ExtendedConsumerProperties<RedisConsumerProperties>, ExtendedProducerProperties<RedisProducerProperties>, RedisProvisioningProvider>
implements ExtendedPropertiesBinder<MessageChannel, RedisConsumerProperties, RedisProducerProperties> {

	private final RedissonClient redissonClient;

	public RedisMessageChannelBinder(String[] headersToEmbed, RedisProvisioningProvider provisioningProvider, RedissonClient redissonClient) {
		super(headersToEmbed, provisioningProvider);
		this.redissonClient = redissonClient;
	}

	@Override
	protected MessageHandler createProducerMessageHandler(ProducerDestination destination, ExtendedProducerProperties<RedisProducerProperties> producerProperties, MessageChannel errorChannel) throws Exception {
		return new RedisMessageHandler(destination.getName(), redissonClient);
	}

	@Override
	protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, ExtendedConsumerProperties<RedisConsumerProperties> properties) throws Exception {
		return new RedisMessageProducerAdapter(destination.getName(), redissonClient);
	}

	@Override
	public RedisConsumerProperties getExtendedConsumerProperties(String channelName) {
		return null;
	}

	@Override
	public RedisProducerProperties getExtendedProducerProperties(String channelName) {
		return null;
	}

	@Override
	public String getDefaultsPrefix() {
		return "";
	}


	@Override
	public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
		return null;
	}
}
